package com.jzlw.cloud.platform.apple.httpsourceexpress;

import com.jzlw.cloud.platform.apple.util.CheckJsonExpress;
import org.apache.commons.lang.StringUtils;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.conf.ConfigurationException;
import org.apache.flume.conf.LogPrivacyUtil;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.source.http.HTTPSourceHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.servlet.http.HttpServletRequest;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.Collections;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.zip.GZIPInputStream;


/**
 * wangyuanhua
 */
public class BlobHandlerExpress implements HTTPSourceHandler {
    public static final String MAX_BLOB_LENGTH_KEY = "maxBlobLength";
    public static final int MAX_BLOB_LENGTH_DEFAULT = 100 * 1000 * 1000;
    private static final int DEFAULT_BUFFER_SIZE = 1024 * 8;
    private static final Logger LOGGER = LoggerFactory.getLogger(BlobHandlerExpress.class);
    private int maxBlobLength = MAX_BLOB_LENGTH_DEFAULT;
    private boolean isCheck = false;
    private int status = 0;
    private String errMsg;
    private int totalAccessedNum = 0;
    private int successAccessedNum = 0;
    private int failAccessedNum = 0;
    private String recordErrMsg;

    public BlobHandlerExpress() {

    }

    public void configure(Context context) {
        this.maxBlobLength = context.getInteger(MAX_BLOB_LENGTH_KEY, MAX_BLOB_LENGTH_DEFAULT);
        this.isCheck = context.getBoolean("IS_CHECK") == null ? false : context.getBoolean("IS_CHECK");
        System.out.println("===========================isCheck:" + isCheck);
        LOGGER.info("===========================isCheck:" + isCheck);
        if (this.maxBlobLength <= 0) {
            throw new ConfigurationException("Configuration parameter " + MAX_BLOB_LENGTH_KEY
                    + " must be greater than zero: " + maxBlobLength);
        }
    }

    //自定义方法，返回错误码
    public Map<String, Object> getMyselfMothed(HttpServletRequest request) {
        List<Event> eventList = null;
        try {
            eventList = this.getEvents(request);
        } catch (Exception e) {
            if (status == 1) {
                LOGGER.error("BlobHandlerMySelf类中解压缩失败!", e);
            } else if (status == 7 || status == 8) {
                LOGGER.error("BlobHandlerMySelf类中验证JSON数据格式失败!", e);
            } else {
                //获取事件集合失败状态码为2
                if (status == 0) {
                    status = 2;
                }
                LOGGER.error("BlobHandlerMySelfOne类中获取事件集合失败!", e);
            }
        }

        //封装返回
        Map<String, Object> retMap = new HashMap();
        retMap.put("status", status);
        retMap.put("errMsg", errMsg);
        retMap.put("eventList", eventList);
        retMap.put("totalAccessedNum", totalAccessedNum);
        retMap.put("successAccessedNum", successAccessedNum);
        retMap.put("failAccessedNum", failAccessedNum);
        if (StringUtils.isNotBlank(recordErrMsg)) {
            retMap.put("recordErrMsg", recordErrMsg);
        }
        return retMap;
    }


    public List<Event> getEvents(HttpServletRequest request) throws Exception {
        Map<String, String> headers = getHeaders(request);
        InputStream in = request.getInputStream();
        try {
            ByteArrayOutputStream blob = null;
            byte[] buf = new byte[Math.min(maxBlobLength, DEFAULT_BUFFER_SIZE)];
            int blobLength = 0;
            int n = 0;
            while ((n = in.read(buf, 0, Math.min(buf.length, maxBlobLength - blobLength))) != -1) {
                if (blob == null) {
                    blob = new ByteArrayOutputStream(n);
                }
                blob.write(buf, 0, n);
                blobLength += n;
                if (blobLength >= maxBlobLength) {
                    LOGGER.warn("Request length exceeds maxBlobLength ({}), truncating BLOB event!",
                            maxBlobLength);
                    break;
                }
            }

            byte[] array = blob != null ? blob.toByteArray() : new byte[0];

            //解压缩
            array = this.uncompress(array);

            //json格式校验
            LOGGER.info("================开始校验收发货订单字段=================");
            CheckJsonExpress checkJson = new CheckJsonExpress();
            if (array == null) {
                LOGGER.error("array为null！无法解析");
                errMsg="array为null！无法解析";
                throw new Exception(errMsg);
            }

            Map inMap = checkJson.validateJsonParams(new String(array, "UTF-8"), isCheck);
            status = (Integer) (inMap.get("status")==null?0:inMap.get("status"));
            errMsg = inMap.get("errMsg") == null ? null : (String) inMap.get("errMsg");
            recordErrMsg = inMap.get("recordErrMsg") == null ? null : (String) inMap.get("recordErrMsg");
            totalAccessedNum = (Integer) inMap.get("total_accessed_num");
            successAccessedNum = (Integer) inMap.get("success_accessed_num");
            //汇总日志
            LOGGER.info("====在BlobHandlerMyselfOne类中校验之后" +
                    ",apple_express_receiver:total_pushed_num:" + totalAccessedNum);
            LOGGER.info("====在BlobHandlerMyselfOne类中校验之后" +
                    ",apple_express_receiver:succeed_pushed_num:" + successAccessedNum);
            LOGGER.info("====在BlobHandlerMyselfOne类中校验之后" +
                    ",apple_express_receiver:failed_pushed_num:" + (totalAccessedNum - successAccessedNum));

            //将解析JSON出现的异常抛出
            if (status == 7 || status == 8 || status == 11 || status == 12 || status == 13 || status == 14) {
                throw new Exception(errMsg);
            }

            //将验证通过的数据继续推送
            if ((totalAccessedNum > successAccessedNum)) {
                String jsonStr = (String) inMap.get("jsonStr");
                array = jsonStr.getBytes(request.getCharacterEncoding() == null ? "UTF-8" : request.getCharacterEncoding());
            }
            LOGGER.info("================校验收发货订单字段结束=================");

            Event event = EventBuilder.withBody(array, headers);
            //Event event = EventBuilder.withBody(new String(array),UTF_8);
            if (LOGGER.isDebugEnabled() && LogPrivacyUtil.allowLogRawData()) {
                LOGGER.debug("blobEvent: {}", event);
            }
            return Collections.singletonList(event);
        } finally {
            in.close();
        }
    }

    private Map<String, String> getHeaders(HttpServletRequest request) {
        if (LOGGER.isDebugEnabled() && LogPrivacyUtil.allowLogRawData()) {
            Map requestHeaders = new HashMap();
            Enumeration iter = request.getHeaderNames();
            while (iter.hasMoreElements()) {
                String name = (String) iter.nextElement();
                requestHeaders.put(name, request.getHeader(name));
            }
            //LOGGER.debug("requestHeaders: {}", requestHeaders);
        }
        Map<String, String> headers = new HashMap();
        if (request.getContentType() != null) {
            headers.put("CONTENT_TYPE", request.getContentType());
        }
        Enumeration iter = request.getParameterNames();
        while (iter.hasMoreElements()) {
            String name = (String) iter.nextElement();
            headers.put(name, request.getParameter(name));
        }
        return headers;
    }

    public byte[] uncompress(byte[] bytes) throws Exception {
        if (bytes == null || bytes.length == 0) {
            return null;
        }
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        ByteArrayInputStream in = new ByteArrayInputStream(bytes);
        try {
            GZIPInputStream ungzip = new GZIPInputStream(in);
            byte[] buffer = new byte[2048];
            int n;
            while ((n = ungzip.read(buffer)) >= 0) {
                out.write(buffer, 0, n);
            }
        } catch (IOException e) {
            //压缩失败的状态码
            status = 1;
            LOGGER.error("BlobHadnlerMySelf类中：gzip uncompress error,解压缩失败", e);
            throw new Exception("BlobHadnlerMySelf类中：gzip uncompress error,解压缩失败", e);
        }
        LOGGER.info("数据解压缩完成");
        return out.toByteArray();
    }
}
